package com.example.flink;

import org.apache.flink.api.common.functions.MapFunction;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;

import lombok.extern.slf4j.Slf4j;

@Slf4j
public class JsonLogParser implements MapFunction<String, LogEvent> {
    private final ObjectMapper mapper = new ObjectMapper();

    @Override
    public LogEvent map(String value) throws Exception {
        JsonNode json = mapper.readTree(value);
        log.info("json: {}", json);
        return new LogEvent(
            json.hasNonNull("timestamp") ? json.get("timestamp").asText() : null,
            json.hasNonNull("level") ? json.get("level").asText() : null,
            json.hasNonNull("message") ? json.get("message").asText() : null,
            json.hasNonNull("thread") ? json.get("thread").asText() : null,
            json.hasNonNull("logger") ? json.get("logger").asText() : null,
            json.hasNonNull("trace_id") ? json.get("trace_id").asText() : null,
            json.hasNonNull("span_id") ? json.get("span_id").asText() : null
        );
    }
} 